package com.liaofan.rocketmq.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class OrderProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("192.168.153.129:9876");

        //若为全局有序,则默认Queue的数量要设为1
        producer.setDefaultTopicQueueNums(1);
        producer.start();

        for (int i = 0; i < 100; i++) {
            //为了演示,使用整形
            Integer orderID = i;
            byte[] body = ("Hi,Liaofan" + i).getBytes();
            Message msg = new Message("someTopic_T1", "someTag_T1", body);
            msg.setKeys("Just Do it:" + i);
            msg.setDelayTimeLevel(3);
            //send的第三个参数会传递给选择器的select()方法的第三个参数
            //该send为同步发送
            SendResult sendResult = producer.send(msg,new MessageQueueSelector() {

                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer id = (Integer) o;
                    int index = id % list.size();

                    return list.get(index);
                }
            }, orderID);
            System.out.println(sendResult);
        }

        producer.shutdown();
    }
}
